"""
Case Type   : 逻辑复制
Case Name   : 备机进行逻辑复制,验证解码数据类型
Create At   : 2021/04/12
Owner       : zopentestcase005
Description :
        1.修改wal_level为logical;enable_slot_log为on
        2.重启数据库
        3.主机pg_hba.conf文件中配置逻辑复制的用户白名单
        4.主机创建逻辑复制槽
        5.主机上查询逻辑复制槽
        6.备机创建解码文件
        7.备机执行逻辑复制槽流式解码
        8.创建表(覆盖基本数据类型)并进行DML操作
        9.备机查看解码文件
        10.停止解码
        11.主机删除逻辑复制槽
        12.清理环境
Expect      :
        1.修改参数wal_level为logical;enable_slot_log为on成功
        2.重启数据库成功
        3.pg_hba.conf 配置逻辑复制的用户白名单成功
        4.主机创建逻辑复制槽成功
        5.显示{self.slot_name}复制槽信息
        6.备机创建解码文件成功
        7.屏幕输出备机逻辑复制槽流式解码过程
        8.创建表(无主键)并进行DML操作成功
        9.解码成功，解码文件update和delete操作不会记录列的旧值(具体哪些数据类型
        不支持，未明确)
        10.停止解码成功
        11.删除成功
        12.清理环境完成
History     :modified by zopentestcase005,2021/10/12,修改用例，适配2.1.0版本
"""
import os
import time
import unittest

from testcase.utils.ComThread import ComThread
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro

Pri_SH = CommonSH('PrimaryDbUser')


@unittest.skipIf(1 == Pri_SH.get_node_num(),
                 '单机环境不执行')
class LogicalReplication(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(
            '----Opengauss_Function_Logical_Replication_Case0010start-----')
        self.constant = Constant()
        self.primary_node = Node('PrimaryDbUser')
        self.standby_node = Node('Standby1DbUser')
        self.standby_node1 = Node('Standby1DbUser')
        self.root_node = Node('Standby1Root')
        self.decode_file = os.path.join(macro.DB_INSTANCE_PATH,
                                        'logical10.txt')
        self.us_name = "us_logical_replication_case0010"
        self.slot_name = "slot_logical_replication_case0010"
        self.tb_name = "tb_logical_replication_case0010"
        self.pg_hba = os.path.join(macro.DB_INSTANCE_PATH,
                                   macro.PG_HBA_FILE_NAME)

    def test_standby_logical(self):
        text = '--step1:修改wal_level为logical;enable_slot_log为on;' \
               'expect:修改成功--'
        self.log.info(text)
        mod_msg = Pri_SH.execute_gsguc('set',
                                       self.constant.GSGUC_SUCCESS_MSG,
                                       'wal_level =logical',
                                       node_name='all',
                                       single=False)
        self.log.info(mod_msg)
        self.assertTrue(mod_msg)
        mod_msg = Pri_SH.execute_gsguc('set',
                                       self.constant.GSGUC_SUCCESS_MSG,
                                       'enable_slot_log =on',
                                       node_name='all',
                                       single=False)
        self.log.info(mod_msg)
        self.assertTrue(mod_msg)

        text = '--step2:重启数据库;expect:重启成功--'
        self.log.info(text)
        restart_msg = Pri_SH.restart_db_cluster()
        self.log.info(restart_msg)
        status = Pri_SH.get_db_cluster_status()
        self.assertTrue("Degraded" in status or "Normal" in status,
                        '执行失败:' + text)

        text = '--step3:配置逻辑复制的用户;expect:配置成功--'
        self.log.info(text)
        sql_cmd = Pri_SH.execut_db_sql(f'''drop role if exists \
            {self.us_name};
            create role {self.us_name} with login password \
            '{macro.COMMON_PASSWD}';
            alter role {self.us_name} with replication sysadmin;''')
        self.log.info(sql_cmd)
        self.assertIn(self.constant.CREATE_ROLE_SUCCESS_MSG, sql_cmd,
                      '执行失败:' + text)
        self.assertIn(self.constant.ALTER_ROLE_SUCCESS_MSG, sql_cmd,
                      '执行失败:' + text)
        self.log.info('--step2.1:配置主机--')
        mod_msg = f"sed -i '$a\local    replication     {self.us_name}      " \
                  f"trust'   {self.pg_hba}"
        self.log.info(mod_msg)
        msg = self.primary_node.sh(mod_msg).result()
        self.log.info(msg)
        mod_msg = f"sed -i '$a\host    replication     " \
                  f"{self.us_name}   127.0.0.1/32   trust'   {self.pg_hba}"
        self.log.info(mod_msg)
        msg = self.primary_node.sh(mod_msg).result()
        self.log.info(msg)
        mod_msg = f"sed -i '$a\host   replication     " \
                  f"{self.us_name}   ::1/128    trust'  {self.pg_hba}"
        self.log.info(mod_msg)
        msg = self.primary_node.sh(mod_msg).result()
        self.log.info(msg)
        self.log.info('--step2.2:配置备机--')
        mod_msg = f"sed -i '$a\local    replication     {self.us_name}     " \
                  f"trust'  {self.pg_hba}"
        self.log.info(mod_msg)
        msg = self.standby_node.sh(mod_msg).result()
        self.log.info(msg)
        mod_msg = f"sed -i '$a\host    replication     {self.us_name}   " \
                  f"127.0.0.1/32   trust'  {self.pg_hba}"
        self.log.info(mod_msg)
        msg = self.standby_node.sh(mod_msg).result()
        self.log.info(msg)
        mod_msg = f"sed -i '$a\host    replication     {self.us_name}   " \
                  f"::1/128   trust'  {self.pg_hba}"
        self.log.info(mod_msg)
        msg = self.standby_node.sh(mod_msg).result()
        self.log.info(msg)
        restart_msg = Pri_SH.restart_db_cluster()
        self.log.info(restart_msg)
        status = Pri_SH.get_db_cluster_status()
        self.assertTrue("Degraded" in status or "Normal" in status,
                        '执行失败:' + text)

        text = '--step4:主机创建逻辑复制槽;expect:创建成功--'
        self.log.info(text)
        check_res = Pri_SH.execut_db_sql("select slot_name from "
                                         "pg_replication_slots;")
        self.log.info(check_res)
        if f'{self.slot_name}' in check_res.split('\n')[-2].strip():
            del_cmd = Pri_SH.execut_db_sql(f"select * from "
                                           f"pg_drop_replication_slot"
                                           f"('{self.slot_name}');")
            self.log.info(del_cmd)
        cre_cmd = Pri_SH.execut_db_sql(f"select * from "
                                       f"pg_create_logical_replication_slot"
                                       f"('{self.slot_name}', "
                                       f"'mppdb_decoding');")
        self.log.info(cre_cmd)

        text = '--step5:查询复制槽;expect:复制槽存在--'
        self.log.info(text)
        query_cmd = Pri_SH.execut_db_sql("select slot_name,plugin from "
                                         "pg_get_replication_slots();")
        self.log.info(query_cmd)
        self.assertIn(f'{self.slot_name}', query_cmd, '执行失败:' + text)

        text = '--step6:备机创建解码文件;expect:创建成功--'
        self.log.info(text)
        touch_cmd = f'''touch {self.decode_file};'''
        self.log.info(touch_cmd)
        result = self.standby_node.sh(touch_cmd).result()
        self.log.info(result)
        self.assertNotIn(self.constant.SQL_WRONG_MSG[1], result,
                         '执行失败:' + text)

        text = '--step7:备机执行逻辑复制槽流式解码;expect:解码无报错--'
        self.log.info(text)
        decode_cmd = f"pg_recvlogical " \
                     f"-d {self.standby_node.db_name} " \
                     f"-S {self.slot_name} " \
                     f"-p {self.standby_node.db_port} " \
                     f"--start " \
                     f"-f {self.decode_file} " \
                     f"-s 2 " \
                     f"-v " \
                     f"-P mppdb_decoding " \
                     f"-U {self.us_name}"
        execute_cmd = f'''source {macro.DB_ENV_PATH}
                           expect <<EOF
                           set timeout 300
                           spawn {decode_cmd}
                           expect "Password:"
                           send "{macro.COMMON_PASSWD}\\n"
                           expect eof\n''' + '''EOF'''
        self.log.info(execute_cmd)
        thread_2 = ComThread(self.standby_node.sh, args=(execute_cmd,))
        thread_2.setDaemon(True)
        thread_2.start()
        thread_2.join(10)
        msg_result_2 = thread_2.get_result()
        self.log.info(msg_result_2)

        text = '--step8:创建表(无主键)并进行DML操作;expect:创建成功--'
        self.log.info(text)
        sql_cmd = Pri_SH.execut_db_sql(f'''drop table if exists \
        {self.tb_name};
        create table {self.tb_name}
        (
        col_tinyint tinyint ,
        col_smallint smallint ,
        col_integer integer,
        col_int int,
        col_binary_integer binary_integer,
        col_bigint bigint,
        col_real real,
        col_float4 float4,
        col_double_precision double precision,
        col_float8 float8,
        col_float float,
        col_float1 float(38),
        col_binary_double binary_double,
        col_char char,
        col_char1 char(50),
        col_character character,
        col_character1 character(50),
        col_varchar varchar,
        col_varchar1 varchar(50),
        col_character_varying character varying(50),
        col_clob clob,
        col_text text,
        col_bytea bytea,
        col_date date,
        col_time time,
        col_time1 time(6),
        col_time2 time without time zone,
        col_time3 time(6) without time zone,
        col_time4 time with time zone,
        col_time5 time(6) with time zone,
        col_timestamp timestamp,
        col_timestamp1 timestamp(6),
        col_timestamp2 timestamp without time zone,
        col_timestamp3 timestamp(6) without time zone,
        col_timestamp4 timestamp with time zone,
        col_timestamp5 timestamp(6) with time zone,
        col_serial serial,
        col_smallserial smallserial,
        col_bigserial bigserial);
        insert into {self.tb_name} (col_tinyint,col_smallint,col_integer,\
        col_int,col_binary_integer,col_bigint,col_real,col_float4,\
        col_double_precision,col_float8,col_float,col_float1,\
        col_binary_double,col_char,col_char1,col_character,col_character1,\
        col_varchar,col_varchar1,col_character_varying,col_clob,col_text,\
        col_bytea,col_date,col_time,col_time1,col_time2,col_time3,col_time4,\
        col_time5,col_timestamp,col_timestamp1,col_timestamp2,col_timestamp3,\
        col_timestamp4,col_timestamp5) \
        values (34,35,36,37,38,39,37.74,75.48,113.22,150.96,188.7,226.44,\
        264.18,'3','wxwhlayyawajbcqzhrctszhddqrwkyzjdwbygz3','h',\
        'V_character_50_length34','V_varchar34', 'V_varchar_5034',\
        'V_character_varying34','V_clob34','V_text34','100100010110',
        '1034-01-08 00:00:00','19:41:34','20:41:35','21:21:36','22:22:37',\
        '21:21:38-08','22:22:39+08','1034-04-22 00:00:00','1035-04-22 pst',
        '1036-04-22 21:22:23','1037-04-22 21:22:23.333333','1038-04-22 pst',\
        '1039-04-22 pst');
        update {self.tb_name} set col_tinyint=0,col_smallint=0,col_integer=0,\
        col_int=0,col_binary_integer=0,col_bigint=0,col_real=0,col_float4=0.0,\
        col_double_precision=0.00,col_float8=0.00,col_float=0.00,\
        col_float1=0.00,col_binary_double=0.00,col_char='0',col_char1='000',\
        col_character='1',col_character1='0000',col_varchar='00003',\
        col_varchar1='00000',col_character_varying='00001',\
        col_clob='v_clob_0',col_text='v_text_0',col_bytea='100101111111',\
        col_date='2000-01-01 00:00:00',col_time='00:00:00',\
        col_time1='00:00:01',col_time2='00:00:02',\
        col_time3='00:00:03',col_time4='00:00:04',col_time5='00:00:05',\
        col_timestamp='2000-1-1',col_timestamp1='2000-1-1 pst',\
        col_timestamp2='2000-1-1 00:00:00',\
        col_timestamp3='2000-1-1 00:00:00.123',col_timestamp4='2000-1-4 pst',\
        col_timestamp5='2000-1-5 pst' where col_tinyint=34;
        delete from {self.tb_name};''')
        self.log.info(sql_cmd)
        self.assertIn(self.constant.TABLE_CREATE_SUCCESS, sql_cmd,
                      '执行失败:' + text)
        time.sleep(3)

        text = '--step9:停止解码;expect:停止解码成功--'
        self.log.info(text)
        stop_cmd = "ps -ef |  grep  pg_recvlogical | grep -v grep | " \
                   "awk '{{print $2}}' | xargs sudo kill -9"
        self.log.info(stop_cmd)
        result = self.root_node.sh(stop_cmd).result()
        self.log.info(result)
        time.sleep(3)

        text = '-step10:查看解码文件;expect:update和delete操作不会记录旧值-'
        self.log.info(text)
        cat_cmd = f"cat {self.decode_file}| grep 'old_keys_name';"
        self.log.info(cat_cmd)
        result = self.standby_node1.sh(cat_cmd).result()
        self.log.info(result)
        self.assertIn('"old_keys_name":[]', result, '执行失败:' + text)

        text = 'step11:主机删除复制槽;expect:删除成功--'
        self.log.info(text)
        del_cmd = Pri_SH.execut_db_sql(f"select * from "
                                       f"pg_drop_replication_slot"
                                       f"('{self.slot_name}');")
        self.log.info(del_cmd)
        self.assertIn('', del_cmd, '执行失败:' + text)

    def tearDown(self):
        text = '--step12:清理环境;expect:清理环境完成--'
        self.log.info(text)
        sql_cmd = Pri_SH.execut_db_sql(f'''drop role if exists {self.us_name};    
                                   drop table if exists {self.tb_name};''')
        self.log.info(sql_cmd)
        rm_cmd = f"rm -rf {self.decode_file};"
        result = self.standby_node.sh(rm_cmd).result()
        self.log.info(result)
        del_msg = f"sed -i '/{self.us_name}/d' {self.pg_hba}"
        self.log.info(del_msg)
        msg = self.standby_node.sh(del_msg).result()
        self.log.info(msg)
        del_msg = f"sed -i '/{self.us_name}/d' {self.pg_hba}"
        self.log.info(del_msg)
        msg = self.primary_node.sh(del_msg).result()
        self.log.info(msg)
        restore_cmd = Pri_SH.execute_gsguc('set',
                                           self.constant.GSGUC_SUCCESS_MSG,
                                           'wal_level=hot_standby')
        self.log.info(restore_cmd)
        restore_cmd = Pri_SH.execute_gsguc('set',
                                           self.constant.GSGUC_SUCCESS_MSG,
                                           'enable_slot_log=off')
        self.log.info(restore_cmd)
        restart_msg = Pri_SH.restart_db_cluster()
        self.log.info(restart_msg)
        status = Pri_SH.get_db_cluster_status()
        self.assertTrue("Degraded" in status or "Normal" in status,
                        '执行失败:' + text)
        self.log.info(
            '--Opengauss_Function_Logical_Replication_Case0010finish----')
